package kafka.examples

import java.util.concurrent.CountDownLatch
import java.util.concurrent.atomic.AtomicBoolean

import grizzled.slf4j.Logger

abstract class ShutdownableThread(val name: String, val isInterruptible: Boolean) extends Thread(name) {
  val logger: Logger = Logger(s"ShutdownableThread[$name]")

  def this(name: String) = {
    this(name, false)
  }

  this.setDaemon(false)

  val isRunning: AtomicBoolean = new AtomicBoolean(true)
  val shutdownLatch: CountDownLatch = new CountDownLatch(1)


  def shutdown(): Unit = {
    logger.info("Shutting down")
    isRunning.set(false)
    if (isInterruptible) {
      interrupt()
    }
    awaitShutdown()
    logger.info("Shutdown completed")
  }

  def doWork(): Unit

  def doCleanJob():Unit


  def awaitShutdown(): Unit = {
    try {
      shutdownLatch.await()
    } catch {
      case e: InterruptedException => throw e
    }
  }

  override def run(): Unit = {
    logger.info("Starting")
    try {
      while (isRunning.get()) {
        doWork()
      }
    } catch {
      case e: Throwable => logger.error("error due to ", e)
    }
    doCleanJob()
    shutdownLatch.countDown()
    logger.info("Stopped ")
  }
}
